package org.dlx.drpc.core.server;

import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

@ChannelHandler.Sharable
public class RPCDispatcher extends ChannelHandlerAdapter
{
    private static Logger logger = LoggerFactory.getLogger(RPCDispatcher.class);
    private ThreadPoolExecutor executor;
    private Map<Class<?>, IRPCHandler> clazzes = new ConcurrentHashMap<>();

    public void register(Class<?> clazz, IRPCHandler handler) {
        clazzes.put(clazz, handler);
    }

    public RPCDispatcher(int workerThreads) {
        BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(1024);
        ThreadFactory factory = new ThreadFactory()
        {
            AtomicInteger seq = new AtomicInteger();
            @Override
            public Thread newThread(Runnable r)
            {
                Thread t = new Thread(r);
                t.setName("rpc-" + seq.getAndIncrement());
                return t;
            }
        };
        this.executor = new ThreadPoolExecutor(1, workerThreads, 30, TimeUnit.SECONDS,
                queue, factory, new ThreadPoolExecutor.CallerRunsPolicy());
    }

    public void closeGracefully() {
        this.executor.shutdown();
        try{
            this.executor.awaitTermination(10, TimeUnit.SECONDS);
        }catch (InterruptedException e){

        }
        this.executor.shutdownNow();
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        // 客户端来了一个新链接
        logger.trace("connection comes");
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        // 客户端走了一个
        logger.trace("connection leaves");
        ctx.close();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        // 此处可能因为客户端机器突发重启
        // 也可能是客户端链接闲置时间超时，后面的ReadTimeoutHandler抛出来的异常
        // 也可能是消息协议错误，序列化异常
        logger.warn("", cause);
        //cause.printStackTrace();
        ctx.close();
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        logger.info("read a message {}", msg);
        // 用业务线程池处理消息
        this.executor.execute(() -> {
            this.dispatch(ctx, msg);
        });
    }

    private void dispatch(ChannelHandlerContext ctx, Object input) {
        try{
            Object pl = input;
            IRPCHandler handler = this.clazzes.get(pl.getClass());
            if(handler != null){
                Object response = handler.handle(pl);
                if(response != null)
                    ctx.writeAndFlush(response);
            }else{
                throw new Exception("不支持的服务");
            }
        }catch (Exception e){
            ctx.close();
        }
    }
}
